/*
Muduo客户端、服务端通信功能的代码实现
*/

#pragma once

#include <iostream>
#include <muduo/net/TcpClient.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <string>
#include <memory>
#include <arpa/inet.h>
#include "Abstract.hpp"
#include "Message.hpp"
#include <mutex>
#include <unordered_map>

namespace RpcModule
{
    // ------------------ MuduoBuffer的实现 ----- 对于muduo缓冲区的封装 ------------------
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf) : _buffer(buf) {};
        // 获取可读数据长度
        virtual size_t ReadableSize() override { return _buffer->readableBytes(); }
        // 尝试获取4字节数据，但不删除
        virtual int32_t PeekInt32() override
        {
            // Muduo库，尝试取出4字节，会自动将网络字节序转换为主机字节序
            // 所以后续就不需要自己再进行一次转换了
            return _buffer->peekInt32();
        }
        // 从缓冲区中删除4字节
        virtual void RetrieveInt32() override { return _buffer->retrieveInt32(); }
        // 从缓冲区中取出4字节，并删除
        virtual int32_t ReadInt32() override { return _buffer->readInt32(); }
        // 获取缓冲区中的len长度数据，返回string
        virtual std::string RetrieveAsString(size_t len) override { return _buffer->retrieveAsString(len); }

    private:
        muduo::net::Buffer *_buffer;
    };

    class MuduoBufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr CreateBuffer(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    // ------------------ LVProtocol的实现 ----- LV格式协议的封装 ------------------
    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;
        // LV格式为：
        // | -- len -- | -- value -- | -- value分为四段，展开为：
        // | -- len -- | -- mtype -- | -- idlen -- | -- id -- | -- body -- |
        // 判断缓冲区中的数据是否完整，完整才可以读取
        virtual bool CanProcessed(const BaseBuffer::ptr &buffer) override
        {
            // peekint32，可能缓冲区中就没有4字节数据，就不能直接取
            if(buffer->ReadableSize() < _lenfieldslength) return false;
            int32_t total_len = buffer->PeekInt32();
            // 如果缓冲区中的可读数据不满足一条完整的数据，返回false
            if (buffer->ReadableSize() < (total_len + _lenfieldslength))
                return false;
            return true;
        }
        // 有消息，从buffer中获取消息，并将消息反序列化到message中
        virtual bool OnMessage(const BaseBuffer::ptr &buffer, BaseMessage::ptr &message) override
        {
            // 这里默认可以直接从缓冲区中获取一条完整的数据
            // 1. 正文长度
            int32_t total_len = buffer->ReadInt32();
            // 2. 消息类型
            MType mtype = (MType)buffer->ReadInt32();
            // 3. id长度
            int32_t id_len = buffer->ReadInt32();
            // 4. 获取id
            std::string id = buffer->RetrieveAsString(id_len);
            // 5. 获取消息正文
            size_t body_len = total_len - _mtypefieldslength - _idlenfieldslength - id_len;
            std::string body = buffer->RetrieveAsString(body_len);
            // 6. 将消息反序列化到message中
            // 6.1 先根据消息类型创建出消息对象
            message = MessageFactory::CreateMessage(mtype);
            if (message.get() == nullptr)
            {
                LOGERROR("消息类型错误，构造消息对象失败！");
                return false;
            }
            bool ret = message->UnSerialize(body);
            if (ret == false)
            {
                LOGERROR("反序列化失败！");
                return false;
            }
            // 7. 注意：收到消息，就要立刻设置消息id和消息类型
            message->SetId(id);
            message->SetMType(mtype);
            return true;
        }
        // 如果要发送数据，需要先序列化 -- LV格式的封装，message中可以进行序列化操作
        virtual std::string Serialize(const BaseMessage::ptr &message) override
        {
            // | -- len -- | -- mtype -- | -- idlen -- | -- id -- | -- body -- |
            std::string body = message->Serialize();
            std::string id = message->GetRid();
            int32_t id_len = htonl(id.size());
            // 注意：需要将主机字节序转换为网络字节序
            auto mtype = htonl((int32_t)message->GetMtype());
            int32_t h_len = _mtypefieldslength + _idlenfieldslength + id.size() + body.size();
            int32_t n_len = htonl(h_len);
            std::string result;
            result.reserve(h_len);
            result.append((char *)&n_len, _lenfieldslength);
            result.append((char *)&mtype, _mtypefieldslength);
            result.append((char *)&id_len, _idlenfieldslength);
            result.append(id);
            result.append(body);
            return result;
        }

    private:
        const size_t _lenfieldslength = 4;
        const size_t _mtypefieldslength = 4;
        const size_t _idlenfieldslength = 4;
    };

    class LVProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr CreateProtocol(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    // ------------------ MuduoConnection的实现 ----- 单个muduo连接的封装 ------------------
    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        // 注意：不需要每次都为一个连接创建一个单独的protocol对象
        // 因为使用的只是一个调用，只需要将一个创建好的protocol传入即可
        MuduoConnection(const BaseProtocol::ptr &protocol,
                        const muduo::net::TcpConnectionPtr &conn) : _protocol(protocol), _conn(conn) {}
        // 发送数据，使用_protocol进行消息的序列化操作
        virtual void Send(const BaseMessage::ptr &message) override
        {
            // 1. 基于LVProtocol，进行序列化操作
            std::string ret = _protocol->Serialize(message);
            // 2. 基于muduo库，进行数据发送
            _conn->send(ret);
        }
        // 关闭连接
        virtual void ShutDown() override
        {
            _conn->shutdown();
        }
        // 判断连接是否正常
        virtual bool Connected() override
        {
            return _conn->connected();
        }

    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class MuduoConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr CreateMuduoConnection(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    // ------------------ MuduoServer的实现 ----- muduo服务端的实现 ------------------
    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port)
            : _server(&_baseloop, muduo::net::InetAddress("127.0.0.1", port), "DictServer", muduo::net::TcpServer::Option::kReusePort),
              _protocol(LVProtocolFactory::CreateProtocol())
        {
            // 设置服务端的回调函数，因为类内成员有this指针，所以需要bind
            _server.setConnectionCallback(std::bind(&MuduoServer::OnConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::OnMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
        }
        void Start()
        {
            // 1. 先开启服务端
            _server.start();
            // 2. 再开启事件循环
            _baseloop.loop();
        }

    private:
        // 连接到来 -- 服务端需要进行连接，同时对连接进行管理
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                std::cout << "建立新连接" << std::endl;
                // 建立新连接，需要对连接进行管理
                BaseConnection::ptr muduo_conn = MuduoConnectionFactory::CreateMuduoConnection(_protocol, conn);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conn));
                }
                // 如果设置了连接到来的回调函数，那么就需要调用回调函数
                if (_cb_connection)
                    _cb_connection(muduo_conn);
            }
            else
            {
                std::cout << "连接断开" << std::endl;
                BaseConnection::ptr muduo_conn;
                {
                    // 连接断开，就取消连接的管理
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto ret = _conns.find(conn);
                    if (ret == _conns.end())
                        return;
                    muduo_conn = ret->second;
                    _conns.erase(conn);
                }
                // 如果设置了连接关闭的回调函数，那么就需要调用回调函数
                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }
        // 消息到来 -- 服务端需要对消息进行处理，但是前提是消息是合法的
        void OnMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            LOGDEBUG("连接有消息到来，开始进行消息的处理！");
            auto base_buffer = MuduoBufferFactory::CreateBuffer(buffer);
            while (1)
            {
                // 1. 如果缓冲区中的数据不完整，不能够进行数据读取
                if (_protocol->CanProcessed(base_buffer) == false)
                {
                    // 2. 如果缓冲区中的数据很大，而且还不完整
                    // 直接断开连接，不允许再发送数据了
                    if (base_buffer->ReadableSize() > maxDataSize)
                    {
                        LOGERROR("缓冲区中的数据太大，连接断开");
                        conn->shutdown();
                        return;
                    }
                    break;
                }
                // 走到这里，缓冲区中有数据，获取一条消息，并进行处理
                BaseMessage::ptr message;
                bool ret = _protocol->OnMessage(base_buffer, message);
                if (ret == false)
                {
                    // 获取消息失败了，直接关闭连接
                    LOGERROR("获取消息失败，关闭连接！");
                    conn->shutdown();
                    return;
                }
                // 然后调用消息到来的回调函数
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    muduo_conn = it->second;
                }
                if (_cb_message)
                    _cb_message(muduo_conn, message);
            }
        }

    private:
        // 2的16次方，64KB
        const size_t maxDataSize = (1 << 16);
        // 应该有一个服务器以及事件监听器
        muduo::net::TcpServer _server;
        muduo::net::EventLoop _baseloop;
        // 判断缓冲区中数据是否完整，需要协议
        BaseProtocol::ptr _protocol;
        // 连接管理
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
        // 访问_conns，需要加锁
        std::mutex _mutex;
    };

    class MuduoServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr CreateMuduoServer(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    // ------------------ MuduoClient的实现 ----- muduo客户端的实现 ------------------
    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &ip, int port)
            : _protocol(LVProtocolFactory::CreateProtocol()),
              _baseloop(_loopthread.startLoop()),
              _downlatch(1),
              _client(_baseloop, muduo::net::InetAddress(ip, port), "DictClient") {}
        // 客户端要连接服务端
        virtual void Connect() override
        {
            LOGDEBUG("客户端正在连接服务器");
            // 1. 注册回调函数
            _client.setConnectionCallback(std::bind(&MuduoClient::OnConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::OnMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
            // 2. 连接服务器，连接之后，才可以停止阻塞，send发送
            _client.connect();
            _downlatch.wait();
            LOGDEBUG("客户端成功连接服务器");
        }
        // 关闭客户端
        virtual void ShutDown() override
        {
            _client.disconnect();
        }
        // 向服务端发送消息
        virtual void Send(const BaseMessage::ptr &message) override
        {
            if (_conn == nullptr || _conn->Connected() == false)
            {
                LOGERROR("客户端已经关闭连接，发送消息失败");
                return;
            }
            _conn->Send(message);
        }
        // 获取当前的连接
        virtual BaseConnection::ptr GetConnetion() override
        {
            return _conn;
        }
        // 检查当前连接是否正常
        virtual bool Connected() override
        {
            return (_conn && (_conn->Connected()));
        }

    private:
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                std::cout << "客户端成功连接上服务端" << std::endl;
                _downlatch.countDown();
                _conn = MuduoConnectionFactory::CreateMuduoConnection(_protocol, conn);
            }
            else
            {
                std::cout << "客户端连接断开" << std::endl;
                _conn.reset();
            }
        }
        void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            LOGDEBUG("服务端有消息到来，开始进行消息的处理！");
            auto base_buffer = MuduoBufferFactory::CreateBuffer(buffer);
            while (1)
            {
                // 1. 如果缓冲区中的数据不完整，不需要读取
                if (_protocol->CanProcessed(base_buffer) == false)
                {
                    // 2. 如果缓冲区中的数据很大，而且还不完整
                    // 直接断开连接，不允许再发送数据了
                    if (base_buffer->ReadableSize() > maxDataSize)
                    {
                        LOGERROR("缓冲区中的数据太大，连接断开");
                        conn->shutdown();
                        return;
                    }
                    break;
                }
                // 走到这里，缓冲区中有数据，获取一条消息，并进行处理
                BaseMessage::ptr message;
                bool ret = _protocol->OnMessage(base_buffer, message);
                if (ret == false)
                {
                    // 获取消息失败了，直接关闭连接
                    LOGERROR("获取消息失败，关闭连接！");
                    conn->shutdown();
                    return;
                }
                // 然后调用消息到来的回调函数
                if (_cb_message)
                    _cb_message(_conn, message);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        // 防止连接还没建立成功就发送消息的情况
        muduo::CountDownLatch _downlatch;
        // 因为事件等待是一个死循环，如果直接死循环，那么客户端只能等待服务端有事件
        // 才可以终止，客户端就无法向服务端发送数据了，所以需要有一个线程专门进行
        // 事件的监听，Muduo库中自己提供了EventLoopThread，会自动创建线程开启监听
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class MuduoClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr CreateMuduoClient(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
};